import {Injectable} from '@angular/core';
import {Observable, Subscription} from 'rxjs/index';
import {Subject} from 'rxjs';

import {QueueingSubject} from '@core/services/websocket/queueing-subject';
import connect from '@core/services/websocket/rxjs-websockets';
import {ConfigService} from '@core/config/config.service';
import {ApiService} from '@core/services/http/api.service';
import {exists} from '../../../utils/utils';
import {map} from 'rxjs/internal/operators';

@Injectable()
export class WebsocketService {
  private messages: Observable<string>;
  private inputStream: QueueingSubject<string>;
  private messagesSubscription: Subscription;
  public messagesObserver: Subject<any>;
  private timer: any;
  private lock: boolean = false;

  constructor(private cs: ConfigService, private api: ApiService) {
    this.messagesObserver = new Subject();
  }

  public connect(userId: string) {
    if (this.messages) {
      if (this.messagesSubscription.closed) {
        this.messagesSubscription = this.messages.subscribe((message: string) => {
          this.messagesObserver.next(message);
        });
      }
      return;
    }

    this.getWsUrl().subscribe(ws_url => {
      if (this.messages) return;
      this.messages = connect(
        ws_url + userId,
        this.inputStream = new QueueingSubject<string>()
      ).messages;

      this.messagesSubscription = this.messages.subscribe((message: string) => {
        this.messagesObserver.next(message);
      });

      if (!this.timer) {
        let that = this;
        this.timer = setInterval(function () {
          that.reconnect(userId);
          that.send('auto Heart Beat!');
        }, 1 * 10 * 1000);
      }
    });
  }

  /**重新连接**/
  private reconnect(userId: string) {
    if (this.lock || this.inputStream == null || this.inputStream.observers.length > 0) return;
    try {
      this.lock = true;
      if (this.messagesSubscription) {
        this.messagesSubscription.unsubscribe();
      }
      if (this.messages) {
        this.messages = null;
      }
      this.connect(userId);
    } finally {
      this.lock = false;
    }
  }

  private getWsUrl(): Observable<any> {
    return this.api.post('ws.getUrl', {}).pipe(map(wsUrl => {
      let ws_url = this.cs.ws_url;
      if (exists(wsUrl)) ws_url = wsUrl;
      return ws_url.endsWith('/') ? ws_url : ws_url + '/';
    }));
  }

  public close() {
    if (this.messagesSubscription) {
      this.messagesSubscription.unsubscribe();
    }
    if (this.messagesObserver) {
      this.messagesObserver.observers = [];
    }
    if (this.messages) {
      this.messages = null;
    }
    if (this.timer) {
      clearInterval(this.timer);
    }
  }

  public send(message: string): void {
    this.inputStream.next(message);
  }

}
